package com.hfzy.ihk.web.aliyuncall.context;

import com.hfzy.ihk.web.aliyuncall.constant.PublicConfig;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.event.AliyunCallOutEvent;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.event.AliyunCallStateEvent;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.event.AliyunRecordFileEvent;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.factory.CallOutEventFactory;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.factory.CallStateEventFactory;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.factory.RecordFileEventFactory;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.handler.AliyunRecordFileHandler;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.handler.asrcall.AliyunAsrCallOutHandler;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.handler.asrcall.AliyunAsrCallStateHandler;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.handler.doublecall.AliyunCallOutHandler;
import com.hfzy.ihk.web.aliyuncall.support.cloudcall.handler.doublecall.AliyunCallStateHandler;
import com.hfzy.ihk.web.aliyuncall.support.disruptor.base.DisruptorEvent;
import com.hfzy.ihk.web.aliyuncall.support.disruptor.exception.DisruptorExceptionHandler;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Administrator on 16-5-28.
 */

@Component("disruptorInitBean")
public class DisruptorInitBean implements InitializingBean, DisposableBean {

    //状态队列
    public static RingBuffer<AliyunCallStateEvent>[] ringBufferAliyunCallState = new RingBuffer[PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE];
    //asr 智能外呼状态队列
    public static RingBuffer<AliyunCallStateEvent>[] ringBufferAliyunAsrCallState = new RingBuffer[PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE];

    //双呼呼出队列
    private static RingBuffer<AliyunCallOutEvent> ringBufferAliyunCallOut;
    //智能外呼呼出队列
    private static RingBuffer<AliyunCallOutEvent> ringBufferAliyunAsrCallOut;
    //接收录音文件队列
    private static RingBuffer<AliyunRecordFileEvent> ringBufferAliyunRecordFile;
    private Logger logger = LoggerFactory.getLogger(DisruptorInitBean.class);
    @Resource
    private TaskExecutor aliyunCallTaskExecutor;
    private RingBuffer<DisruptorEvent> ringBufferBatchInsert;
    private Disruptor<DisruptorEvent> disruptorBatchInsert;
    private ExecutorService executorBatchInsert;


    private Disruptor<AliyunCallOutEvent> disruptorAliyunCallOut;

    private Disruptor<AliyunCallOutEvent> disruptorAliyunAsrCallOut;

    private Disruptor<AliyunCallStateEvent> disruptorAliyunCallState;

    private Disruptor<AliyunCallStateEvent> disruptorAliyunAsrCallState;

    private Disruptor<AliyunRecordFileEvent> disruptorAliyunRecordFile;


    //线程
    private ExecutorService executorFreeswitchCallOut;
    private ExecutorService executorFreeswitchCallState;
    private ExecutorService executorAliyunCallOut;
    private ExecutorService executorAliyunAsrCallOut;
    private ExecutorService executorAliyunCallState;
    private ExecutorService executorAliyunAsrCallState;
    private ExecutorService executorAliyunRecordFile;

    public static RingBuffer<AliyunCallOutEvent> getRingBufferAliyunCallOut() {
        return ringBufferAliyunCallOut;
    }

    public static RingBuffer<AliyunCallOutEvent> getRingBufferAliyunAsrCallOut() {
        return ringBufferAliyunAsrCallOut;
    }

    public static RingBuffer<AliyunCallStateEvent> getRingBufferAliyunCallState(int num) {
        return ringBufferAliyunCallState[num];
    }

    public static RingBuffer<AliyunCallStateEvent> getRingBufferAliyunAsrCallState(int num) {
        return ringBufferAliyunAsrCallState[num];
    }

    public static RingBuffer<AliyunRecordFileEvent> getRingBufferAliyunRecordFile() {
        return ringBufferAliyunRecordFile;
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        logger.info("开始初始化云呼队列");

        //双呼
        initDisruptorBatchInsert();//必须先初始化
        initDisruptorAliyunCallOut();
        initDisruptorAliyunCallState();
        initDisruptorAliyunRecordFile();
        //asr智能外呼
        initDisruptorAliyunAsrCallOut();
        initDisruptorAliyunAsrCallState();
        logger.info("初始化云呼队列完成");
    }



    /**
     * 初始化双呼呼出队列
     */
    private void initDisruptorAliyunCallOut() {
        int handlerSize = PublicConfig.ALIYUN_CALL_OUT_HANDLER_SIZE;
        int bufferSize = PublicConfig.ALIYUN_CALL_OUT_BUFFER_SIZE;

        executorAliyunCallOut = Executors.newFixedThreadPool(handlerSize);

        CallOutEventFactory callEventFactory = new CallOutEventFactory();

        disruptorAliyunCallOut = new Disruptor(
                callEventFactory,
                bufferSize,
                executorAliyunCallOut,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        AliyunCallOutHandler handlers[] = new AliyunCallOutHandler[handlerSize];
        for (int i = 0; i < handlerSize; i++) {
            handlers[i] = BeanUtils.getAliyunCallOutHandler();
        }

        disruptorAliyunCallOut.handleEventsWithWorkerPool(handlers);

        disruptorAliyunCallOut.handleExceptionsWith(new DisruptorExceptionHandler());

        ringBufferAliyunCallOut = disruptorAliyunCallOut.start();

        logger.info("init disruptorAliyunCallOut over: handler size:{},buffer size:{}", handlerSize, bufferSize);
    }

    /**
     * 初始化智能外呼呼出队列
     */
    private void initDisruptorAliyunAsrCallOut() {
        int handlerSize = PublicConfig.ALIYUN_CALL_OUT_HANDLER_SIZE;
        int bufferSize = PublicConfig.ALIYUN_CALL_OUT_BUFFER_SIZE;

        executorAliyunAsrCallOut = Executors.newFixedThreadPool(handlerSize);

        CallOutEventFactory callEventFactory = new CallOutEventFactory();

        disruptorAliyunAsrCallOut = new Disruptor(
                callEventFactory,
                bufferSize,
                executorAliyunAsrCallOut,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        AliyunAsrCallOutHandler handlers[] = new AliyunAsrCallOutHandler[handlerSize];
        for (int i = 0; i < handlerSize; i++) {
            handlers[i] = BeanUtils.getAliyunAsrCallOutHandler();
        }

        disruptorAliyunAsrCallOut.handleEventsWithWorkerPool(handlers);

        disruptorAliyunAsrCallOut.handleExceptionsWith(new DisruptorExceptionHandler());

        ringBufferAliyunAsrCallOut = disruptorAliyunAsrCallOut.start();

        logger.info("init disruptorAliyunAsrCallOut over: handler size:{},buffer size:{}", handlerSize, bufferSize);
    }




    /**
     * 初始化中间状态队列 这里包括中间状态和账单 （双呼）
     */
    private void initDisruptorAliyunCallState() {
        int handlerSize = PublicConfig.ALIYUN_CALL_STATE_HANDLER_SIZE;
        int bufferSize = PublicConfig.ALIYUN_CALL_STATE_BUFFER_SIZE;

        int bufferNumSize = PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE;

        for (int j = 0; j < bufferNumSize; j++) {
            executorAliyunCallState = Executors.newFixedThreadPool(handlerSize);

            CallStateEventFactory callEventFactory = new CallStateEventFactory();

            disruptorAliyunCallState = new Disruptor(
                    callEventFactory,
                    bufferSize,
                    executorAliyunCallState,
                    ProducerType.MULTI,
                    new BlockingWaitStrategy()
            );

            AliyunCallStateHandler handlers[] = new AliyunCallStateHandler[handlerSize];
            for (int i = 0; i < handlerSize; i++) {
                handlers[i] = BeanUtils.getAliyunCallStateHandler();
            }

            disruptorAliyunCallState.handleEventsWithWorkerPool(handlers);

            disruptorAliyunCallState.handleExceptionsWith(new DisruptorExceptionHandler());
            ringBufferAliyunCallState[j] = disruptorAliyunCallState.start();
        }


        logger.info("init disruptorAliyunCallState over: handler size:{},buffer size:{}", handlerSize, bufferSize);
    }



    /**
     * 初始化中间状态队列 这里包括中间状态和账单（ASR智能外呼）
     */
    private void initDisruptorAliyunAsrCallState() {
        int handlerSize = PublicConfig.ALIYUN_CALL_STATE_HANDLER_SIZE;
        int bufferSize = PublicConfig.ALIYUN_CALL_STATE_BUFFER_SIZE;

        int bufferNumSize = PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE;

        for (int j = 0; j < bufferNumSize; j++) {

            executorAliyunAsrCallState = Executors.newFixedThreadPool(handlerSize);

            CallStateEventFactory callEventFactory = new CallStateEventFactory();

            disruptorAliyunAsrCallState = new Disruptor(
                    callEventFactory,
                    bufferSize,
                    executorAliyunAsrCallState,
                    ProducerType.MULTI,
                    new BlockingWaitStrategy()
            );

            AliyunAsrCallStateHandler handlers[] = new AliyunAsrCallStateHandler[handlerSize];
            for (int i = 0; i < handlerSize; i++) {
                handlers[i] = BeanUtils.getAliyunAsrCallStateHandler();
            }

            disruptorAliyunAsrCallState.handleEventsWithWorkerPool(handlers);

            disruptorAliyunAsrCallState.handleExceptionsWith(new DisruptorExceptionHandler());
            ringBufferAliyunAsrCallState[j] = disruptorAliyunAsrCallState.start();
        }

        logger.info("init disruptorAliyunAsrCallState over: handler size:{},buffer size:{}", handlerSize, bufferSize);
    }

    /**
     * 初始化录音文件队列
     */
    private void initDisruptorAliyunRecordFile() {
        int handlerSize = PublicConfig.ALIYUN_RECORD_FILE_HANDLER_SIZE;
        int bufferSize = PublicConfig.ALIYUN_RECORD_FILE_BUFFER_SIZE;

        executorAliyunRecordFile = Executors.newFixedThreadPool(handlerSize);

        RecordFileEventFactory callEventFactory = new RecordFileEventFactory();

        disruptorAliyunRecordFile = new Disruptor(
                callEventFactory,
                bufferSize,
                executorAliyunRecordFile,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        AliyunRecordFileHandler handlers[] = new AliyunRecordFileHandler[handlerSize];
        for (int i = 0; i < handlerSize; i++) {
            handlers[i] = BeanUtils.getAliyunRecordFileHandler();
        }

        disruptorAliyunRecordFile.handleEventsWithWorkerPool(handlers);

        disruptorAliyunRecordFile.handleExceptionsWith(new DisruptorExceptionHandler());

        ringBufferAliyunRecordFile = disruptorAliyunRecordFile.start();

        logger.info("init disruptorAliyunRecordFile over: handler size:{},buffer size:{}", handlerSize, bufferSize);
    }



    /*单消费者*/
    private void initDisruptorBatchInsert() {

//        int handlerSize = PublicConfig.PUBLIC_BATCH_INSERT_HANDLER_SIZE;
//        int bufferSize  = PublicConfig.PUBLIC_BATCH_INSERT_BUFFER_SIZE;
//
//        executorBatchInsert = Executors.newFixedThreadPool(handlerSize);
//
//        CallOutEventFactory callEventFactory = new CallOutEventFactory();
//
//        disruptorBatchInsert = new Disruptor(
//                callEventFactory,
//                bufferSize,
//                executorBatchInsert,
//                ProducerType.MULTI,
//                new BlockingWaitStrategy()
//        );
//
//        // Connect the handler
//        BatchInsertHandler handlers[] = new BatchInsertHandler[handlerSize];
//        for(int i=0;i<handlerSize;i++){
//            handlers[i] = new BatchInsertHandler("catchInsertHandler"+i);
//        }
//
//        disruptorBatchInsert.handleEventsWith(handlers);//.then(new KickAssEventHandler());
//
//        //同一个事件，先用handler1处理再用handler2处理。
////        disruptorBatchInsert.handleEventsWith(c1).then(handler2);
//        // Start the Disruptor, starts all threads running
//        ringBufferBatchInsert = disruptorBatchInsert.start();
//
//        logger.info("init disruptorBatchInsert over: handler size:{},buffer size:{}",handlerSize,bufferSize);
    }

    @Override
    public void destroy() throws Exception {
        logger.info("disruptor destroy..............");
        //================================
        disruptorBatchInsert.shutdown();
        executorBatchInsert.shutdown();

        disruptorAliyunCallOut.shutdown();
        disruptorAliyunAsrCallOut.shutdown();

        disruptorAliyunCallState.shutdown();
        disruptorAliyunAsrCallState.shutdown();
        disruptorAliyunRecordFile.shutdown();

        executorAliyunCallOut.shutdown();
        executorAliyunCallState.shutdown();
        executorAliyunRecordFile.shutdown();

        //关闭freeswitch相关
        executorFreeswitchCallState.shutdown();

        executorFreeswitchCallOut.shutdown();
    }

}
